/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.reading;

import java.nio.ByteBuffer;
import java.util.Comparator;

import edu.ucsd.db.bassarisk.data.api.IValueType;
import edu.ucsd.db.bassarisk.data.api.IValueType.ITypeTools;
import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;
import edu.ucsd.db.bassarisk.storage.common.IColumnSegmentInfo;
import edu.ucsd.db.bassarisk.storage.reading.IDrawerIndexReader.CardInfo;

/**
 * Serves as a base class for various CardReader classes that decompress
 * sequences of values from columns in the Bassarisk column store database.
 * 
 * Each different CardReader implementation decompresses a different value type,
 * compression scheme combination.
 * 
 * @author nbales
 */
public abstract class AbstractCardReader<VT extends IValueType> implements
    ICardReader<VT> {

  protected class Cursor {
    int byteOffset;
    long index;
  }

  /**
   * A reference to the ByteBuffer which contains the postings currently
   * represented by this object.
   */
  protected ByteBuffer drawerBuffer;

  protected final Cursor current;
  
  protected Comparator<IValueReference<VT>> valueComparator;
  
  protected boolean initialized;

  /**
   * Information about the card in the current buffer.
   */
  protected CardInfo<VT> cardInfo;

  public AbstractCardReader(final ITypeTools<VT> columnTypeTools) {
    this.current = new Cursor();

    this.cardInfo = null;

    // FIXME: This is the slow way to compare, we can use fast non-generic 
    //        implementations once the query engine is working.
    this.valueComparator = columnTypeTools.getGenericComparator();
  }
  
  @Override
  public boolean isInitialized() {
    return this.initialized;
  }
  
  @Override
  public long currentIndex() {
    return this.current.index - 1; // current.index stays one step ahead of current value
  }

  @Override
  public void close() {
    // nothing to do yet
  }

  @Override
  public boolean hasEntryIndex(final long targetEntryIndex) {
    if (this.cardInfo == null) { return false; }
    long firstIndex = this.cardInfo.getFirstEntryIndex();
    long numEntries = this.cardInfo.getEntryCount();

    return targetEntryIndex >= firstIndex
        && targetEntryIndex < (firstIndex + numEntries);
  }
  
  @Override
  public boolean hasValue(IValueReference<VT> targetValue) {
    if (this.cardInfo == null) { return false; }
    IValueReference<VT> firstValue = this.cardInfo.getFirstValue();
    IValueReference<VT> lastValue = this.cardInfo.getLastValue();
    
    return this.valueComparator.compare(targetValue, firstValue) >= 0
        && this.valueComparator.compare(targetValue, lastValue) <= 0;
  }

  @Override
  public void resetCard(final CardInfo<VT> cardInfo) {
    this.cardInfo = cardInfo;

    this.current.byteOffset =
        this.cardInfo.getCardIndex() * ColumnStore.BYTES_IN_CARD;
    this.current.index = this.cardInfo.getFirstEntryIndex();
  }

  @Override
  public void resetDrawer(final ByteBuffer drawerBuffer,
                          final IColumnSegmentInfo<VT> drawerInfo) {
    this.drawerBuffer = drawerBuffer;
  }

  protected boolean sanityCheckByte(final int byteIndex) {
    int cardIndex = this.cardInfo.getCardIndex();
    return byteIndex >= cardIndex * ColumnStore.BYTES_IN_CARD
        && byteIndex < (cardIndex + 1) * ColumnStore.BYTES_IN_CARD;
  }

}